package own.stu.jobgib.playown.io.theadPool;


import org.apache.commons.lang3.time.DateFormatUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import own.stu.jobgib.playown.io.common.TServer;
import own.stu.jobgib.playown.io.common.TServerSocket;
import own.stu.jobgib.playown.io.common.TSocket;

import java.io.IOException;
import java.util.Date;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

public class TheadPoolServer extends TServer {

    private static final Logger LOGGER = LoggerFactory.getLogger(TheadPoolServer.class);

    public static void main(String[] args) throws Exception {
        TheadPoolServer server = new TheadPoolServer(new Args().minWorkerThreads(2).maxWorkerThreads(2).port(8888));
        System.out.println("server started ........ ");
        server.serve();
    }

    public static class Args {
        public int minWorkerThreads = 5;
        public int maxWorkerThreads = Integer.MAX_VALUE;
        public ExecutorService executorService;
        public int stopTimeoutVal = 5;
        public TimeUnit stopTimeoutUnit = TimeUnit.SECONDS;
        public int port;

        public Args minWorkerThreads(int n) {
            minWorkerThreads = n;
            return this;
        }

        public Args maxWorkerThreads(int n) {
            maxWorkerThreads = n;
            return this;
        }

        public Args stopTimeoutVal(int n) {
            stopTimeoutVal = n;
            return this;
        }

        public Args stopTimeoutUnit(TimeUnit tu) {
            stopTimeoutUnit = tu;
            return this;
        }

        public Args port(int p) {
            port = p;
            return this;
        }

        public Args executorService(ExecutorService executorService) {
            this.executorService = executorService;
            return this;
        }
    }

    // executor service for handing client connections
    private ExecutorService executorService_;

    private final TimeUnit stopTimeoutUnit;

    private final long stopTimeoutVal;

    /**
     * Timeout for client sockets from accept
     */
    private int clientTimeout_ = 0;

    public TheadPoolServer(Args args) throws Exception {
        stopTimeoutUnit = args.stopTimeoutUnit;
        stopTimeoutVal = args.stopTimeoutVal;

        executorService_ = args.executorService != null ? args.executorService : createDefaultExecutorService(args);
        serverSocket = new TServerSocket(args.port);
    }

    public void serve() {
        if (!preServe()) {
            return;
        }

        execute();

        executorService_.shutdownNow();

        if (!waitForShutdown()) {
            LOGGER.error("Shutdown is not done after " + stopTimeoutVal + stopTimeoutUnit);
        }

        setServing(false);
    }

    protected void execute() {
        while (!stopped_) {
            try {
                TSocket socket = serverSocket.accept();
                try {
                    executorService_.execute(new WorkerProcess(socket));
                } catch (RejectedExecutionException e) {
                    if (!stopped_) {
                        LOGGER.warn("ThreadPool is saturated with incoming requests. Closing latest connection.");
                    }
                    socket.close();
                }
            } catch (Exception e) {
                if (!stopped_) {
                    LOGGER.warn("Transport error occurred during acceptance of message", e);
                }
            }
        }
    }

    protected boolean waitForShutdown() {

        long timeoutMS = stopTimeoutUnit.toMillis(stopTimeoutVal);
        long now = System.currentTimeMillis();
        while (timeoutMS >= 0) {
            try {
                return executorService_.awaitTermination(timeoutMS, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                long newNow = System.currentTimeMillis();
                timeoutMS -= (newNow - now);
                now = newNow;
            }
        }
        return false;
    }

    protected boolean preServe() {
        try {
            serverSocket.listen();
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        stopped_ = false;
        setServing(true);
        return true;
    }

    protected ExecutorService getExecutorService() {
        return executorService_;
    }

    private static ExecutorService createDefaultExecutorService(Args args) {
        return new ThreadPoolExecutor(args.minWorkerThreads, args.maxWorkerThreads, 60L, TimeUnit.SECONDS,
                new SynchronousQueue<>(), new ThreadFactory() {
            final AtomicLong count = new AtomicLong();

            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setDaemon(true);
                thread.setName(String.format("TThreadPoolServer WorkerProcess-%d", count.getAndIncrement()));
                return thread;
            }
        });
    }

    private class WorkerProcess implements Runnable {

        TSocket socket;

        public WorkerProcess(TSocket socket) {
            this.socket = socket;
        }

        @Override
        public void run() {
            try {
                while (true) {
                    if (Thread.currentThread().isInterrupted()) {
                        LOGGER.debug("WorkerProcess requested to shutdown");
                        break;
                    }

                    try {

                        // String s = StreamUtils.copyToString(inputStream, Charset.defaultCharset());
                        System.out.println("from client : " + socket.read());
                    } catch (IOException e) {
                        e.printStackTrace();
                        throw new RuntimeException("get inputStream's data error, ", e);
                    }

                    try {
                        //StreamUtils.copy("hello ... " + new Date(), Charset.defaultCharset(), socket.getOutputStream_());
                        // socket.getOutputStream_().flush();
                        socket.write(DateFormatUtils.format(new Date(), "yyyy-MM-dd hh:mm:ss") + " : " + "hale");
                    } catch (IOException e) {
                        e.printStackTrace();
                        throw new RuntimeException("write outputStream_'s data error, ", e);
                    }
                }
            } catch (Exception e) {
                LOGGER.debug("Error processing request", e);
            } finally {
                socket.close();
            }
        }
    }
}
